package com.luych.toolbox.data.service.job;

import com.luych.toolbox.common.enums.DataCompareHistoryStatus;
import com.luych.toolbox.common.enums.DataComparePlanStatus;
import com.luych.toolbox.data.service.constant.Constant;
import com.luych.toolbox.data.service.dao.DataCompareHistoryDao;
import com.luych.toolbox.data.service.dao.DataComparePlanDao;
import com.luych.toolbox.data.service.entity.DataCompareHistory;
import com.luych.toolbox.data.service.entity.DataComparePlan;
import org.quartz.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.quartz.QuartzJobBean;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

public class DataComparePlanExecuteJob extends QuartzJobBean {
    private static final Logger LOGGER = LoggerFactory.getLogger(DataComparePlanExecuteJob.class);

    private final Scheduler scheduler;

    private final DataComparePlanDao dataComparePlanDao;
    private final DataCompareHistoryDao dataCompareHistoryDao;

    public DataComparePlanExecuteJob(Scheduler scheduler, DataComparePlanDao dataComparePlanDao,
                                     DataCompareHistoryDao dataCompareHistoryDao) {
        this.scheduler = scheduler;
        this.dataComparePlanDao = dataComparePlanDao;
        this.dataCompareHistoryDao = dataCompareHistoryDao;
    }

    private JobDetail getJobDetail(String id) {
        Map<String, Object> jobParam = new HashMap<String, Object>() {{
            this.put("id", id);
        }};
        return JobBuilder.newJob(DataCompareHistoryExecuteJob.class)
                .withIdentity(id, DataCompareHistoryExecuteJob.class.getName())
                .usingJobData(new JobDataMap(jobParam))
                .build();
    }

    private Trigger getJobTrigger(String id) {
        return TriggerBuilder.newTrigger()
                .withIdentity(id, DataCompareHistoryExecuteJob.class.getName())
                .withSchedule(SimpleScheduleBuilder.repeatSecondlyForTotalCount(1, 1))
                .build();
    }

    @Override
    protected void executeInternal(JobExecutionContext context) throws JobExecutionException {
        String id = context.getJobDetail().getJobDataMap().getString("id");

        DataComparePlan dataComparePlan = dataComparePlanDao.findById(id).orElseThrow(() ->
                new JobExecutionException(Constant.DATA_COMPARE_PLAN_NOT_FOUND));

        if (DataComparePlanStatus.DISABLE.equals(dataComparePlan.getStatus())) {
            LOGGER.warn("the data compare plan[{}] is disabled, but still run in quartz, why ???", dataComparePlan.getId());
            return;
        }

        DataCompareHistory compareHistory = new DataCompareHistory();
        compareHistory.setId(UUID.randomUUID().toString());
        compareHistory.setCreateUserId(dataComparePlan.getCreateUserId());
        compareHistory.setCreateDate(new Date());
        compareHistory.setName(dataComparePlan.getName());
        compareHistory.setStatus(DataCompareHistoryStatus.LOADING);
        compareHistory.setDataSourceOneId(dataComparePlan.getDataSourceOneId());
        compareHistory.setDataSourceTwoId(dataComparePlan.getDataSourceTwoId());
        compareHistory.setDataComparePlanId(dataComparePlan.getId());
        compareHistory.setOwnerId(dataComparePlan.getOwnerId());
        dataCompareHistoryDao.save(compareHistory);

        LOGGER.info("start to schedule data compare plan [{}]...", dataComparePlan.getName());
        try {
            scheduler.scheduleJob(this.getJobDetail(compareHistory.getId()), this.getJobTrigger(compareHistory.getId()));
            LOGGER.info("success to schedule data compare plan [{}]", dataComparePlan.getName());
        } catch (SchedulerException e) {
            LOGGER.info("error to schedule data compare plan [{}]: {}", dataComparePlan.getName(), e.getMessage(), e);
            throw new JobExecutionException(e);
        }
    }
}
